package com.z2.disruptor;

import com.z2.disruptor.common.EventProcessType;
import com.z2.disruptor.comsumer.LongBatchEventHandler;
import com.z2.disruptor.comsumer.LongEventPoolHandler;
import com.z2.disruptor.event.LongEvent;
import com.z2.disruptor.event.LongEventFactory;
import com.z2.disruptor.producer.LongEventProducerWithTranslator;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

/**
 * Created by cdyizuoming on 2017/11/9.
 */
public class TestMain {

    private static Integer CORE_SIZE = 4;
    private static Integer BUFFER_SIZE = 4;
    private static Integer EVENT_TYPE = 2;
    private static Disruptor<LongEvent> disruptor;


    public static void main(String[] args) throws Exception {
        init();
        //设置事件业务处理器---消费者
        comsumerSetUp(EVENT_TYPE);
        //启动disruptor线程
        disruptor.start();
        // 获取 ring buffer环，用于接取生产者生产的事件
        RingBuffer<LongEvent> ringBuffer = disruptor.getRingBuffer();
        LongEventProducerWithTranslator producer = new LongEventProducerWithTranslator(ringBuffer);
        new Thread(new TestTask(producer, 10)).start();
    }


    /**
     * 消费者设置
     *
     * @param eventType
     * @return
     */
    public static void comsumerSetUp(Integer eventType) {
        if (EventProcessType.BATCH_EVENT_PROCESSOR.getCode().equals(eventType)) {
            LongBatchEventHandler handler = new LongBatchEventHandler();
            disruptor.handleEventsWith(handler, handler, handler, handler);
        } else if (EventProcessType.WORK_EVENT_PROCESSOR.getCode().equals(eventType)) {
            LongEventPoolHandler handler = new LongEventPoolHandler();
            disruptor.handleEventsWithWorkerPool(handler, handler, handler, handler);
        }
    }


    /**
     * 初始化
     */
    public static void init() {
        // 执行器，用于构造消费者线程
        Executor executor = Executors.newFixedThreadPool(CORE_SIZE);
        // 指定事件工厂
        LongEventFactory factory = new LongEventFactory();
        // 指定 ring buffer字节大小, must be power of 2.
        int bufferSize = BUFFER_SIZE;
        //单线程模式，获取额外的性能
        disruptor = new Disruptor<LongEvent>(factory,
                bufferSize, executor,
                ProducerType.SINGLE,
                new BlockingWaitStrategy());
    }


}




